ATOM Documentation

┌─────────────────────────────────────────────────────────────────┐
│                         Frontend Layer                          │
├─────────────────────────────────────────────────────────────────┤
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │ Template     │  │ Workflow     │  │ Node         │          │
│  │ Browser      │→ │ Builder      │→ │ Config       │          │
│  │ (207-01)     │  │ (207-02)     │  │ Sidebar      │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
│         ↓                   ↓                   ↓               │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │ Integration  │  │ Variable     │  │ Template     │          │
│  │ Node         │  │ Substitution │  │ Save         │          │
│  │ Selector     │  │ Panel        │  │ (207-08)     │          │
│  │ (207-03)     │  │ (207-07)     │  │              │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
└─────────────────────────────────────────────────────────────────┘
                              ↓ HTTPS
┌─────────────────────────────────────────────────────────────────┐
│                         API Layer                               │
├─────────────────────────────────────────────────────────────────┤
│  GET /api/workflows-templates         (Template listing)        │
│  GET /api/workflows-templates/{id}    (Template fetch)          │
│  GET /api/integrations               (Capability discovery)     │
│  GET /api/entities/types             (Entity types)             │
│  GET /api/agents                     (Agent listing)            │
│  POST /api/workflows/execute         (Workflow execution)       │
│  POST /api/workflows/save-template   (Save template)            │
└─────────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────────┐
│                      Backend Services Layer                     │
├─────────────────────────────────────────────────────────────────┤
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │ Workflow     │  │ Integration  │  │ Entity       │          │
│  │ Template     │  │ Registry     │  │ Skill        │          │
│  │ Manager      │  │ (201-02)     │  │ Executor     │          │
│  │ (206-01)     │  │              │  │ (204-03)     │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
│         ↓                   ↓                   ↓               │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │ Unified      │  │ Agent        │  │ Outflow      │          │
│  │ Action       │  │ Governance   │  │ Wrapper      │          │
│  │ Executor     │  │ Service      │  │ Factory      │          │
│  │ (202-01)     │  │ (204-01)     │  │ (203-01)     │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
└─────────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────────┐
│                      Data Layer                                 │
├─────────────────────────────────────────────────────────────────┤
│  workflow_templates/    (File-based template storage)           │
│  TenantIntegration      (Tenant integration configurations)      │
│  OAuthToken            (OAuth credentials per tenant)           │
│  Entity                (Custom entity types)                    │
│  Agent                 (AI agents with maturity levels)         │
└─────────────────────────────────────────────────────────────────┘

interface WorkflowBuilderProps {
  onSave?: (data: { nodes: Node[]; edges: Edge[] }) => void;
  initialData?: { nodes: Node[]; edges: Edge[] };
  initialTemplateId?: string; // Load template on mount
}

const [nodes, setNodes, onNodesChange] = useNodesState(initialNodes);
const [edges, setEdges, onEdgesChange] = useEdgesState(initialEdges);
const [variableValues, setVariableValues] = useState<Record<string, any>>({});
const [templateInputs, setTemplateInputs] = useState<TemplateInput[]>([]);

interface TemplateBrowserProps {
  onLoadTemplate: (templateId: string) => void;
  onClose?: () => void;
}

const fetchTemplates = async (category?: string, search?: string) => {
  const params = new URLSearchParams();
  if (category && category !== "All") params.append("category", category);
  if (search) params.append("search", search);

  const response = await fetch(`/api/workflows-templates?${params}`);
  const data = await response.json();
  return data.templates;
};

{
  id: string;
  type: "integration";
  position: {
    x: number;
    y: number;
  }
  data: {
    label: string;
    connector_id: string; // e.g., 'slack', 'salesforce'
    operation: string; // e.g., 'send_message', 'create_lead'
    parameters: Array<{ name: string; value: any }>;
    connection_status: "connected" | "not_connected";
  }
}

{
  id: string;
  type: "entity";
  position: {
    x: number;
    y: number;
  }
  data: {
    label: string;
    entity_type: string; // e.g., 'Lead', 'Task', 'Deal'
    operation: "create" | "read" | "update" | "delete" | "query";
    parameters: Record<string, any>;
    maturity_level: "student" | "intern" | "supervised" | "autonomous";
  }
}

{
  id: string;
  type: 'ai_node';
  position: { x: number; y: number };
  data: {
    label: string;
    agent_id: string;
    agent_role: string;
    maturity_level: 'student' | 'intern' | 'supervised' | 'autonomous';
    prompt: string;
    governance_restrictions: string[];
  };
}

class WorkflowTemplateManager:
    def load_templates(self) -> None:
        """Load all templates from workflow_templates/ directory."""

    def get_template(self, template_id: str) -> Optional[WorkflowTemplate]:
        """Get template by ID."""

    def instantiate_template(self, template_id: str, inputs: Dict[str, Any]) -> WorkflowInstance:
        """Instantiate template with input values."""

    def save_template(self, template: WorkflowTemplate, tenant_id: str) -> str:
        """Save template to workflow_templates/ directory."""

    def reload_templates(self) -> int:
        """Reload all templates from disk."""

class IntegrationRegistry:
    def get_service(self, tenant_id: str, connector_id: str) -> IntegrationService:
        """Get integration service instance for tenant."""

    def discover_capabilities(self, connector_id: str) -> Dict[str, Any]:
        """Discover available operations for integration."""

    def resolve_credentials(self, tenant_id: str, connector_id: str) -> Dict[str, Any]:
        """Resolve credentials for integration."""

class UnifiedActionExecutor:
    async def dispatch(self, tenant_id: str, step: WorkflowStep, context: Dict[str, Any]) -> ActionResult:
        """Dispatch workflow step to appropriate executor."""

    async def dispatch_integration_action(self, tenant_id: str, connector_id: str, action: str, params: Dict[str, Any]) -> ActionResult:
        """Dispatch integration action via IntegrationRegistry."""

    async def dispatch_entity_operation(self, tenant_id: str, entity_type: str, operation: str, params: Dict[str, Any]) -> ActionResult:
        """Dispatch entity operation via EntitySkillExecutor."""

    async def dispatch_agent_step(self, tenant_id: str, agent_id: str, step: WorkflowStep, context: Dict[str, Any]) -> ActionResult:
        """Dispatch agent step via reasoning engine."""

class EntitySkillExecutor:
    async def create_entity(self, tenant_id: str, entity_type: str, data: Dict[str, Any], maturity_level: str) -> Entity:
        """Create entity instance."""

    async def read_entity(self, tenant_id: str, entity_type: str, entity_id: str, maturity_level: str) -> Entity:
        """Read entity instance."""

    async def update_entity(self, tenant_id: str, entity_type: str, entity_id: str, data: Dict[str, Any], maturity_level: str) -> Entity:
        """Update entity instance."""

    async def delete_entity(self, tenant_id: str, entity_type: str, entity_id: str, maturity_level: str) -> None:
        """Delete entity instance."""

    async def query_entities(self, tenant_id: str, entity_type: str, filters: Dict[str, Any], maturity_level: str) -> List[Entity]:
        """Query entities with filters."""

User clicks template in TemplateBrowser
  ↓
TemplateBrowser.onLoadTemplate(templateId)
  ↓
WorkflowBuilder.handleLoadTemplate(templateId)
  ↓
GET /api/workflows-templates/{templateId}
  ↓
WorkflowTemplateManager.get_template(templateId)
  ↓
Template returned as JSON
  ↓
templateToReactFlow(template) in template-loader.ts
  ↓
Generate ReactFlow nodes and edges
  ↓
setNodes(), setEdges() in WorkflowBuilder
  ↓
Template displayed on canvas

User clicks "Execute Workflow" button
  ↓
WorkflowBuilder.handleExecuteWorkflow()
  ↓
nodesToWorkflowSteps(nodes) in executor.ts
  ↓
POST /api/workflows/execute
  ↓
UnifiedActionExecutor.dispatch() for each step
  ↓
Governance check (AgentGovernanceService)
  ↓
Quota check (QuotaManagerService)
  ↓
Route to executor (Integration/Entity/Agent)
  ↓
Execute with retry logic
  ↓
Return results
  ↓
formatExecutionResults(results)
  ↓
Display results in UI

User clicks "Save as Template" button
  ↓
WorkflowBuilder.handleSaveAsTemplate()
  ↓
Prompt for template metadata (name, description, category)
  ↓
reactFlowToTemplate(nodes, edges, metadata)
  ↓
POST /api/workflows/save-template
  ↓
WorkflowTemplateManager.save_template()
  ↓
Validate template schema
  ↓
Save to workflow_templates/{category}-{name}-{version}.json
  ↓
Return success message

export function templateToReactFlow(template: WorkflowTemplate): {
  nodes: Node[];
  edges: Edge[];
  complexity: TemplateComplexity;
};

const nodeTypeMap: Record<string, string> = {
  trigger: "trigger",
  agent_execution: "ai_node",
  integration_action: "action",
  conditional_logic: "condition",
  entity_operation: "entity",
  loop: "loop",
};

const nodeType = nodeTypeMap[step.step_type] || "action";

function generateEdges(steps: WorkflowStep[]): Edge[] {
  const edges: Edge[] = [];
  const stepIndexMap = new Map<string, number>();

  // Build step index map
  steps.forEach((step, index) => {
    stepIndexMap.set(step.step_id, index);
  });

  // Generate edges from dependencies
  steps.forEach((step, index) => {
    if (step.depends_on) {
      step.depends_on.forEach((depId) => {
        const depIndex = stepIndexMap.get(depId);
        if (depIndex !== undefined) {
          edges.push({
            id: `e${depIndex}-${index}`,
            source: depIndex.toString(),
            target: index.toString(),
            type: "addStepEdge",
          });
        }
      });
    }
  });

  return edges;
}

interface Edge {
  id: string;
  source: string; // Source node ID
  target: string; // Target node ID
  type?: string; // Edge type (default, addStepEdge, etc.)
  label?: string; // Optional label
  animated?: boolean; // Animation flag
  style?: React.CSSProperties; // Custom styles
}

function generateNodePositions(
  steps: WorkflowStep[],
): Map<string, { x: number; y: number }> {
  const positions = new Map<string, { x: number; y: number }>();
  const visited = new Set<string>();
  const depths = new Map<string, number>();

  // Calculate depths using DFS
  function calculateDepth(stepId: string, depth: number): void {
    depths.set(stepId, depth);
    const step = steps.find((s) => s.step_id === stepId);
    if (step?.depends_on) {
      step.depends_on.forEach((depId) => {
        if (!visited.has(depId)) {
          visited.add(depId);
          calculateDepth(depId, depth + 1);
        }
      });
    }
  }

  // Start from root nodes (no dependencies)
  steps.forEach((step) => {
    if (!step.depends_on || step.depends_on.length === 0) {
      calculateDepth(step.step_id, 0);
    }
  });

  // Group steps by depth
  const depthGroups = new Map<number, string[]>();
  depths.forEach((depth, stepId) => {
    if (!depthGroups.has(depth)) {
      depthGroups.set(depth, []);
    }
    depthGroups.get(depth)!.push(stepId);
  });

  // Calculate positions
  const HORIZONTAL_SPACING = 300;
  const VERTICAL_SPACING = 150;

  depthGroups.forEach((stepIds, depth) => {
    const y = depth * VERTICAL_SPACING;
    const width = (stepIds.length - 1) * HORIZONTAL_SPACING;
    const startX = -width / 2;

    stepIds.forEach((stepId, index) => {
      const x = startX + index * HORIZONTAL_SPACING;
      positions.set(stepId, { x, y });
    });
  });

  return positions;
}

interface IntegrationNodeData {
  label: string; // Display name (e.g., "Send Slack Message")
  connector_id: string; // Integration identifier (e.g., "slack")
  operation: string; // Operation name (e.g., "send_message")
  parameters: Array<{
    name: string; // Parameter name
    value: any; // Parameter value (can include variable references)
    type: string; // Parameter type (string, number, boolean, etc.)
    required: boolean; // Whether parameter is required
  }>;
  connection_status: "connected" | "not_connected"; // OAuth connection status
}

{
  id: 'node-1',
  type: 'integration',
  position: { x: 100, y: 100 },
  data: {
    label: 'Send Slack Message',
    connector_id: 'slack',
    operation: 'send_message',
    parameters: [
      { name: 'channel', value: '${input.slack_channel}', type: 'string', required: true },
      { name: 'message', value: '${step_1.result}', type: 'string', required: true },
    ],
    connection_status: 'connected',
  },
}

interface EntityNodeData {
  label: string; // Display name (e.g., "Create Lead")
  entity_type: string; // Entity type name (e.g., "Lead", "Task", "Deal")
  operation: "create" | "read" | "update" | "delete" | "query"; // CRUD operation
  parameters: Record<string, any>; // Entity data
  maturity_level: "student" | "intern" | "supervised" | "autonomous"; // Required maturity
  maturity_color: string; // Color code based on maturity
}

{
  id: 'node-2',
  type: 'entity',
  position: { x: 100, y: 300 },
  data: {
    label: 'Create Lead',
    entity_type: 'Lead',
    operation: 'create',
    parameters: {
      first_name: '${input.first_name}',
      last_name: '${input.last_name}',
      email: '${input.email}',
      status: 'new',
    },
    maturity_level: 'intern',
    maturity_color: '#3b82f6', // Blue
  },
}

interface AgentNodeData {
  label: string; // Display name (e.g., "Analyze Lead")
  agent_id: string; // Agent identifier
  agent_role: string; // Agent role (e.g., "Sales Assistant")
  maturity_level: "student" | "intern" | "supervised" | "autonomous"; // Agent maturity
  prompt: string; // Task prompt for agent
  governance_restrictions: string[]; // List of governance restrictions
  can_execute: boolean; // Whether agent can execute operation
}

{
  id: 'node-3',
  type: 'ai_node',
  position: { x: 100, y: 500 },
  data: {
    label: 'Analyze Lead',
    agent_id: 'agent-123',
    agent_role: 'Sales Assistant',
    maturity_level: 'intern',
    prompt: 'Analyze this lead and determine if it qualifies for follow-up',
    governance_restrictions: [
      'Approval required for actions',
      'Cannot delete records',
      'Limited to read operations',
    ],
    can_execute: true,
  },
}

interface TriggerNodeData {
  label: string; // Display name
  trigger_type: "webhook" | "schedule" | "manual"; // Trigger type
  webhook_url?: string; // Webhook URL (if webhook trigger)
  schedule?: string; // Cron expression (if schedule trigger)
  schema?: any; // Input schema for webhook payload
}

{
  id: 'node-0',
  type: 'trigger',
  position: { x: 100, y: 0 },
  data: {
    label: 'Webhook Start',
    trigger_type: 'webhook',
    webhook_url: 'https://atomagentos.com/api/webhooks/wh-123',
    schema: {
      type: 'object',
      properties: {
        userId: { type: 'string' },
        action: { type: 'string' },
      },
    },
  },
}

interface ActionNodeData {
  label: string; // Display name
  action_type: string; // Action type
  parameters: Record<string, any>; // Action parameters
}

{
  id: 'node-4',
  type: 'action',
  position: { x: 100, y: 700 },
  data: {
    label: 'Send Email',
    action_type: 'send_email',
    parameters: {
      to: '${input.email}',
      subject: 'Workflow Complete',
      body: 'Your workflow has completed successfully.',
    },
  },
}

interface ConditionNodeData {
  label: string; // Display name
  condition: string; // Condition expression
  true_label: string; // Label for true branch
  false_label: string; // Label for false branch
}

{
  id: 'node-5',
  type: 'condition',
  position: { x: 100, y: 400 },
  data: {
    label: 'Is Lead Qualified?',
    condition: '${step_1.score} > 80',
    true_label: 'Yes',
    false_label: 'No',
  },
}

{
  "step_id": "send_notification",
  "parameters": [
    {
      "name": "message",
      "default_value": "Workflow '${input.workflow_name}' completed with status: ${step_1.status}"
    }
  ]
}

export function extractVariables(template: WorkflowTemplate): {
  inputs: TemplateInput[];
  variables: VariableReference[];
};

const VARIABLE_PATTERN = /\$\{(?:input|step_\d+|step_[\w-]+)\.[\w.]+\}/g;

def substitute_variables(value: str, inputs: Dict[str, Any], context: Dict[str, Any]) -> Any:
    """Substitute variables in parameter value."""

    if not isinstance(value, str):
        return value

    # Replace ${input.field} variables
    value = re.sub(r'\$\{input\.([\w.]+)\}', lambda m: str(inputs.get(m.group(1), '')), value)

    # Replace ${step_N.field} variables
    value = re.sub(r'\$\{step_(\d+)\.([\w.]+)\}', lambda m: str(context.get(int(m.group(1)), {}).get(m.group(2), '')), value)

    # Replace ${step_step_id.field} variables
    value = re.sub(r'\$\{step_([\w-]+)\.([\w.]+)\}', lambda m: str(context.get(m.group(1), {}).get(m.group(2), '')), value)

    return value

interface ValidationError {
  type:
    | "invalid_source"
    | "invalid_field_path"
    | "invalid_step_reference"
    | "input_not_found"
    | "step_not_found";
  message: string;
  variable: string;
  location: string; // Step ID or parameter name
}

GET /api/workflows-templates?category=integration&search=salesforce
Authorization: Bearer {session_token}

{
  "templates": [
    {
      "template_id": "salesforce-slack-sync-1.0.0",
      "name": "Salesforce to Slack Sync",
      "description": "Sync new Salesforce leads to Slack channel",
      "category": "integration",
      "complexity": "intermediate",
      "inputs": [
        {
          "name": "slack_channel",
          "label": "Slack Channel",
          "type": "string",
          "required": true,
          "default_value": "#sales"
        }
      ],
      "steps": [
        {
          "step_id": "query_leads",
          "name": "Query Salesforce Leads",
          "step_type": "integration_action",
          "service": "salesforce",
          "action": "query_leads",
          "parameters": [{ "name": "limit", "default_value": "10" }]
        },
        {
          "step_id": "send_slack",
          "name": "Send to Slack",
          "step_type": "integration_action",
          "service": "slack",
          "action": "send_message",
          "depends_on": ["query_leads"],
          "parameters": [
            { "name": "channel", "default_value": "${input.slack_channel}" },
            {
              "name": "message",
              "default_value": "New leads: ${step_query_leads.count}"
            }
          ]
        }
      ],
      "is_public": true,
      "tenant_id": null,
      "tags": ["salesforce", "slack", "sync"]
    }
  ],
  "total": 1
}

@router.get("/workflows-templates")
async def browse_templates(
    category: Optional[str] = None,
    search: Optional[str] = None,
    session: Session = Depends(get_db),
    tenant: Tenant = Depends(get_tenant)
):
    manager = WorkflowTemplateManager(session)
    templates = manager.browse_templates(
        tenant_id=tenant.id,
        category=category,
        search=search
    )
    return {"templates": templates, "total": len(templates)}

GET /api/workflows-templates/salesforce-slack-sync-1.0.0
Authorization: Bearer {session_token}

GET /api/integrations
Authorization: Bearer {session_token}

{
  "integrations": [
    {
      "connector_id": "slack",
      "name": "Slack",
      "description": "Send messages and channels",
      "icon": "slack-icon.png",
      "connection_status": "connected",
      "operations": [
        {
          "name": "send_message",
          "description": "Send message to channel",
          "parameters": [
            { "name": "channel", "type": "string", "required": true },
            { "name": "message", "type": "string", "required": true }
          ]
        }
      ]
    }
  ]
}

@router.get("/integrations")
async def discover_integrations(
    session: Session = Depends(get_db),
    tenant: Tenant = Depends(get_tenant)
):
    registry = IntegrationRegistry(session)
    integrations = registry.discover_all(tenant_id=tenant.id)
    return {"integrations": integrations}

GET /api/entities/types
Authorization: Bearer {session_token}

{
  "entity_types": [
    {
      "name": "Lead",
      "schema": {
        "first_name": { "type": "string" },
        "last_name": { "type": "string" },
        "email": { "type": "string" },
        "status": {
          "type": "string",
          "enum": ["new", "contacted", "qualified"]
        }
      },
      "maturity_level": "intern"
    }
  ]
}

GET /api/agents
Authorization: Bearer {session_token}

{
  "agents": [
    {
      "id": "agent-123",
      "name": "Sales Assistant",
      "role": "Sales Assistant",
      "maturity_level": "intern",
      "capabilities": ["analyze_leads", "send_followup"]
    }
  ]
}

POST /api/workflows/execute
Authorization: Bearer {session_token}
Content-Type: application/json

{
  "steps": [
    {
      "step_id": "query_leads",
      "step_type": "integration_action",
      "connector_id": "salesforce",
      "operation": "query_leads",
      "parameters": { "limit": 10 }
    },
    {
      "step_id": "send_slack",
      "step_type": "integration_action",
      "connector_id": "slack",
      "operation": "send_message",
      "depends_on": ["query_leads"],
      "parameters": {
        "channel": "${input.slack_channel}",
        "message": "New leads: ${step_query_leads.count}"
      }
    }
  ],
  "inputs": {
    "slack_channel": "#sales"
  }
}

{
  "execution_id": "exec-456",
  "status": "completed",
  "results": [
    {
      "step_id": "query_leads",
      "status": "success",
      "output": { "count": 5, "leads": [...] }
    },
    {
      "step_id": "send_slack",
      "status": "success",
      "output": { "message_ts": "123456.789" }
    }
  ],
  "errors": []
}

{
  "execution_id": "exec-456",
  "status": "failed",
  "results": [],
  "errors": [
    {
      "step_id": "send_slack",
      "error": "Integration not connected",
      "code": "INTEGRATION_NOT_CONNECTED"
    }
  ]
}

@router.post("/workflows/execute")
async def execute_workflow(
    request: WorkflowExecutionRequest,
    session: Session = Depends(get_db),
    tenant: Tenant = Depends(get_tenant)
):
    executor = UnifiedActionExecutor(session)
    execution_id = str(uuid.uuid4())

    results = []
    errors = []
    context = {}

    for step in request.steps:
        try:
            result = await executor.dispatch(
                tenant_id=tenant.id,
                step=step,
                context=context
            )
            results.append(result)
            context[step.step_id] = result.output
        except Exception as e:
            errors.append({
                "step_id": step.step_id,
                "error": str(e),
                "code": getattr(e, "code", "UNKNOWN_ERROR")
            })

    return {
        "execution_id": execution_id,
        "status": "completed" if not errors else "failed",
        "results": results,
        "errors": errors
    }

POST /api/workflows/save-template
Authorization: Bearer {session_token}
Content-Type: application/json

{
  "name": "My Custom Workflow",
  "description": "A custom workflow for my use case",
  "category": "automation",
  "complexity": "intermediate",
  "steps": [...],
  "is_public": false
}

{
  "template_id": "automation-my-custom-workflow-1.0.0",
  "status": "saved",
  "message": "Template saved successfully"
}

@router.post("/workflows/save-template")
async def save_template(
    template_data: WorkflowTemplateCreate,
    session: Session = Depends(get_db),
    tenant: Tenant = Depends(get_tenant)
):
    manager = WorkflowTemplateManager(session)
    template_id = await manager.save_template(
        template=template_data,
        tenant_id=tenant.id
    )
    return {
        "template_id": template_id,
        "status": "saved",
        "message": "Template saved successfully"
    }

def browse_templates(tenant_id: str, category: Optional[str] = None) -> List[WorkflowTemplate]:
    query = self.session.query(WorkflowTemplate).filter(
        or_(
            WorkflowTemplate.is_public == True,
            WorkflowTemplate.tenant_id == tenant_id
        )
    )

    if category:
        query = query.filter(WorkflowTemplate.category == category)

    return query.all()

def get_service(self, tenant_id: str, connector_id: str) -> IntegrationService:
    # Fetch tenant's integration configuration
    tenant_integration = self.session.query(TenantIntegration).filter(
        TenantIntegration.tenant_id == tenant_id,
        TenantIntegration.connector_id == connector_id
    ).first()

    if not tenant_integration:
        raise IntegrationNotConfigured()

    # Resolve credentials for tenant
    credentials = self.resolve_credentials(tenant_id, connector_id)

    # Load service with credentials
    service = self._load_service(connector_id)
    service.configure(credentials)

    return service

async def query_entities(self, tenant_id: str, entity_type: str, filters: Dict[str, Any]) -> List[Entity]:
    # Validate entity type exists for tenant
    entity_def = self.session.query(EntityDefinition).filter(
        EntityDefinition.tenant_id == tenant_id,
        EntityDefinition.name == entity_type
    ).first()

    if not entity_def:
        raise EntityNotFound()

    # Query entities with tenant filter
    query = self.session.query(Entity).filter(
        Entity.tenant_id == tenant_id,
        Entity.entity_type == entity_type
    )

    # Apply filters
    for key, value in filters.items():
        query = query.filter(getattr(Entity, key) == value)

    return query.all()

async def get_agents(self, tenant_id: str) -> List[Agent]:
    agents = self.session.query(Agent).filter(
        Agent.tenant_id == tenant_id
    ).all()

    return [
        {
            "id": agent.id,
            "name": agent.name,
            "role": agent.role,
            "maturity_level": agent.maturity_level,
            "capabilities": agent.capabilities
        }
        for agent in agents
    ]

export function nodesToWorkflowSteps(nodes: Node[]): WorkflowStep[] {
  return nodes.map((node, index) => {
    const step: WorkflowStep = {
      step_id: node.id,
      step_type: mapNodeTypeToStepType(node.type),
      name: node.data.label,
    };

    // Add connector_id for integration nodes
    if (node.type === "integration") {
      step.connector_id = node.data.connector_id;
      step.operation = node.data.operation;
      step.parameters = node.data.parameters;
    }

    // Add entity_type for entity nodes
    if (node.type === "entity") {
      step.entity_type = node.data.entity_type;
      step.operation = node.data.operation;
      step.parameters = node.data.parameters;
    }

    // Add agent_id for agent nodes
    if (node.type === "ai_node") {
      step.agent_id = node.data.agent_id;
      step.prompt = node.data.prompt;
    }

    // Add depends_on from edges
    const incomingEdges = edges.filter((e) => e.target === node.id);
    if (incomingEdges.length > 0) {
      step.depends_on = incomingEdges.map((e) => e.source);
    }

    return step;
  });
}

function mapNodeTypeToStepType(nodeType: string): string {
  const mapping = {
    trigger: "trigger",
    ai_node: "agent_execution",
    integration: "integration_action",
    entity: "entity_operation",
    condition: "conditional_logic",
    action: "action",
  };
  return mapping[nodeType] || "action";
}

async def dispatch(self, tenant_id: str, step: WorkflowStep, context: Dict[str, Any]) -> ActionResult:
    """Dispatch workflow step to appropriate executor."""

    # 1. Validate step structure
    self._validate_step(step)

    # 2. Check governance
    if step.step_type == 'agent_execution':
        decision = await self.governance_service.can_perform_action(
            tenant_id=tenant_id,
            agent_id=step.agent_id,
            action=step.operation
        )
        if not decision.allowed:
            raise GovernanceBlocked(decision.reason)

    # 3. Check quota
    await self.quota_service.check_quota(tenant_id, 'workflow_execution')

    # 4. Route to executor
    if step.step_type == 'integration_action':
        result = await self.dispatch_integration_action(
            tenant_id=tenant_id,
            connector_id=step.connector_id,
            action=step.operation,
            params=step.parameters,
            context=context
        )
    elif step.step_type == 'entity_operation':
        result = await self.dispatch_entity_operation(
            tenant_id=tenant_id,
            entity_type=step.entity_type,
            operation=step.operation,
            params=step.parameters,
            context=context
        )
    elif step.step_type == 'agent_execution':
        result = await self.dispatch_agent_step(
            tenant_id=tenant_id,
            agent_id=step.agent_id,
            step=step,
            context=context
        )
    else:
        raise UnknownStepType(step.step_type)

    # 5. Log action
    await self._log_action_result(tenant_id, step, result)

    # 6. Return result
    return result

async def can_perform_action(self, tenant_id: str, agent_id: str, action: str) -> GovernanceDecision:
    """Check if agent can perform action."""

    # Get agent
    agent = self.session.query(Agent).filter(
        Agent.tenant_id == tenant_id,
        Agent.id == agent_id
    ).first()

    if not agent:
        raise AgentNotFound()

    # Check maturity-based permissions
    if agent.maturity_level == 'student' and action in DESTRUCTIVE_ACTIONS:
        return GovernanceDecision(
            allowed=False,
            reason="Student agents cannot perform destructive actions"
        )

    if agent.maturity_level == 'intern' and action in RESTRICTED_ACTIONS:
        return GovernanceDecision(
            allowed=False,
            reason="Intern agents require approval for this action"
        )

    return GovernanceDecision(allowed=True)

async def dispatch_integration_action(
    self,
    tenant_id: str,
    connector_id: str,
    action: str,
    params: Dict[str, Any],
    context: Dict[str, Any]
) -> ActionResult:
    """Dispatch integration action."""

    # Get service from registry
    service = await self.registry.get_service(tenant_id, connector_id)

    # Substitute variables in parameters
    resolved_params = self._substitute_variables(params, context)

    # Execute operation with retry
    result = await self._execute_with_retry(
        lambda: service.execute_operation(action, resolved_params)
    )

    return ActionResult(
        status="success",
        output=result
    )

async def create_entity(self, tenant_id: str, entity_type: str, data: Dict[str, Any], maturity_level: str) -> Entity:
    """Create entity instance."""

    # Validate maturity permissions
    if maturity_level == 'student':
        raise PermissionDenied("Student agents cannot create entities")

    # Validate entity schema
    entity_def = self._validate_entity_type(tenant_id, entity_type)

    # Create entity
    entity = Entity(
        tenant_id=tenant_id,
        entity_type=entity_type,
        data=data
    )
    self.session.add(entity)
    self.session.commit()

    return entity

const handleExecuteWorkflow = async () => {
  const steps = nodesToWorkflowSteps(nodes);

  const response = await fetch("/api/workflows/execute", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      steps,
      inputs: variableValues,
    }),
  });

  const result = await response.json();

  if (result.status === "completed") {
    toast({
      title: "Workflow executed successfully",
      description: `Completed ${result.results.length} steps`,
    });
  } else {
    toast({
      title: "Workflow execution failed",
      description: result.errors.map((e) => e.error).join(", "),
      variant: "destructive",
    });
  }
};

{
  "error": "Governance blocked",
  "code": "GOVERNANCE_BLOCKED",
  "message": "Student agents cannot perform destructive actions",
  "step_id": "delete_record",
  "status": 403
}

{
  "error": "Quota exceeded",
  "code": "QUOTA_EXCEEDED",
  "message": "Daily workflow execution quota exceeded (500/500)",
  "quota_limit": 500,
  "quota_used": 500,
  "quota_reset": "2026-03-29T00:00:00Z",
  "status": 429
}

{
  "error": "Integration not connected",
  "code": "INTEGRATION_NOT_CONNECTED",
  "message": "Slack integration not connected. Please connect your Slack account.",
  "connector_id": "slack",
  "connect_url": "https://atomagentos.com/api/integrations/slack/connect",
  "status": 400
}

{
  "error": "Internal server error",
  "code": "INTERNAL_ERROR",
  "message": "An unexpected error occurred. Please try again later.",
  "request_id": "req-123456",
  "status": 500
}

const getErrorMessage = (error: any): string => {
  const errorMessages = {
    GOVERNANCE_BLOCKED:
      "Your agent's maturity level does not permit this action.",
    QUOTA_EXCEEDED: "You've reached your daily workflow execution limit.",
    INTEGRATION_NOT_CONNECTED: `Please connect your ${error.connector_id} account.`,
    INTERNAL_ERROR: "Something went wrong. Please try again later.",
  };

  return errorMessages[error.code] || error.message;
};

def test_template_tenant_isolation():
    # Create template for tenant A
    template_a = create_template(tenant_id="tenant-a", is_public=False)

    # Fetch templates for tenant B
    templates_b = browse_templates(tenant_id="tenant-b")

    # Assert tenant A's template not visible to tenant B
    assert template_a.id not in [t.id for t in templates_b]

def test_credential_tenant_isolation():
    # Configure Slack for tenant A
    configure_integration(tenant_id="tenant-a", connector_id="slack", token="token-a")

    # Try to execute integration as tenant B
    with pytest.raises(IntegrationNotConfigured):
        execute_integration(tenant_id="tenant-b", connector_id="slack")

def test_entity_tenant_isolation():
    # Create entity for tenant A
    entity_a = create_entity(tenant_id="tenant-a", entity_type="Lead", data={"name": "Lead A"})

    # Query entities for tenant B
    entities_b = query_entities(tenant_id="tenant-b", entity_type="Lead")

    # Assert tenant A's entity not visible to tenant B
    assert entity_a.id not in [e.id for e in entities_b]

async def test_load_and_execute_template():
    # 1. Browse templates
    templates = await browse_templates(category="integration")
    assert len(templates) > 0

    # 2. Load template into builder
    template = templates[0]
    nodes, edges = template_to_reactflow(template)

    # 3. Execute workflow
    result = await execute_workflow(nodes, edges, inputs={})
    assert result["status"] == "completed"

async def test_create_integration_node():
    # 1. Fetch integrations
    integrations = await discover_integrations()
    slack = next(i for i in integrations if i["connector_id"] == "slack")

    # 2. Create integration node
    node = create_integration_node(
        connector_id="slack",
        operation="send_message",
        parameters={"channel": "#test", "message": "Hello"}
    )

    # 3. Verify node structure
    assert node["type"] == "integration"
    assert node["data"]["connector_id"] == "slack"
    assert node["data"]["operation"] == "send_message"

async def test_variable_substitution():
    # 1. Create template with variables
    template = {
        "inputs": [{"name": "name", "type": "string"}],
        "steps": [
            {
                "step_id": "greet",
                "operation": "send_message",
                "parameters": {
                    "message": "Hello, ${input.name}!"
                }
            }
        ]
    }

    # 2. Execute with input
    result = await execute_workflow(
        template,
        inputs={"name": "Alice"}
    )

    # 3. Verify variable substituted
    assert result["steps"][0]["output"]["message"] == "Hello, Alice!"